package cn._51doit.live.jobs;

import cn._51doit.live.utils.FlinkUtils;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 使用FlinkSQL对累计观众和在线观众进行统计，写入到MySQL中
 */
public class OnlineUserCountV3 {

    public static void main(String[] args) throws Exception {

        ParameterTool parameterTool = ParameterTool.fromPropertiesFile(args[0]);
        StreamTableEnvironment tEnv = FlinkUtils.getStreamTableEnv();
        FlinkUtils.env.enableCheckpointing(parameterTool.getInt("checkpoint.interval", 60000));

        //定义一个Kafka的Source
        //使用KafkaConnector从Kafka中读取数据
        tEnv.executeSql(
                "CREATE TABLE tb_user_event (\n" +
                        "  `carrier` STRING,  \n" +
                        "  `deviceId` STRING,  \n" +
                        "  `deviceType` STRING,  \n" +
                        "  `eventId` STRING,  \n" +
                        "  `isNew` INT,  \n" +
                        "  `latitude` DOUBLE,  \n" +
                        "  `longitude` DOUBLE,  \n" +
                        "  `netType` STRING,  \n" +
                        "  `osName` STRING,  \n" +
                        "  `osVersion` STRING,  \n" +
                        "  `properties` MAP<STRING, STRING>,  \n" +
                        "  `releaseChannel` STRING,  \n" +
                        "  `resolution` STRING,  \n" +
                        "  `sessionId` STRING,  \n" +
                        "  `timestamp` BIGINT  \n" +
                        ") WITH (\n" +
                        "  'connector' = 'kafka',\n" +
                        "  'topic' = 'user-log-28',\n" +
                        "  'properties.bootstrap.servers' = 'node-1.51doit.cn:9092,node-2.51doit.cn:9092,node-3.51doit.cn:9092',\n" +
                        "  'properties.group.id' = 'testGroup',\n" +
                        "  'scan.startup.mode' = 'earliest-offset',\n" +
                        "  'format' = 'json', \n" +
                        "  'json.ignore-parse-errors' = 'true'" +
                        ")"
        );



        //从source中读取数据，并处理后插入到sink表中

        //多维度统计新用户的（统计插入到Mysql的tb_new_user）
        //tEnv.executeSql("insert into tb_print_table select * from tb_user_event where eventId = 'appLaunch'");

        //统计在线用户和累计用户
        TableResult tableResult = tEnv.executeSql("select channel, os, sum(flag1) online_count, sum(flag2) total_count from (select releaseChannel channel, osName os, if(eventId = 'liveEnter', 1, -1) flag1, if(eventId = 'liveEnter', 1, 0) flag2 , `properties`['anchor_id'] anchor_id from tb_user_event where eventId = 'liveEnter' or eventId = 'liveLeave') group by cube (channel, os)");

        tableResult.print();


    }

}
